package org.example.mq;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;

/**
 * @author: wf
 * @create: 2022/8/30 15:54
 * @description:
 */
public class ProducerService {
    private static final Logger logger = LoggerFactory.getLogger(ProducerService.class);
    private DefaultMQProducer producer = null;
    
    public ProducerService() {
    }
    
    public ProducerService(String nameSrvAddress, String groupName, int retryTimes) {
        logger.info("ProducerService initProducer,nameSrvAddress:{},groupName:{},retryTimes:{}", new Object[]{nameSrvAddress, groupName, retryTimes});
        this.producer = new DefaultMQProducer(groupName);
        this.producer.setNamesrvAddr(nameSrvAddress);
        this.producer.setRetryTimesWhenSendFailed(retryTimes);
        
        try {
            this.producer.start();
        } catch (MQClientException var5) {
            logger.error("ProducerService initProducer error,messge:{}", var5.getErrorMessage());
        }
        
    }
    
    public void send(String topic, String msg) {
        this.send(topic,"", msg);
    }
    
    public void send(String topic, String key, String msg) {
        this.send(topic, key, "", msg);
    }
    
    public void send(String topic, String key, String tags, String msg) {
        try {
            Message message = new Message(topic, tags, key, msg.getBytes(StandardCharsets.UTF_8));
            this.producer.send(message);
        } catch (MQClientException var6) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{}", new Object[]{var6.getMessage(), topic, key, msg});
        } catch (InterruptedException var7) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{}", new Object[]{var7.getMessage(), topic, key, msg});
        } catch (RemotingException var8) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{}", new Object[]{var8.getMessage(), topic, key, msg});
        } catch (MQBrokerException var9) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{}", new Object[]{var9.getMessage(), topic, key, msg});
        }
        
    }
    
    /**
     * 发送延迟消息
     * @param topic
     * @param key
     * @param msg
     * @param delayTimeLevel
     */
    public void send(String topic, String key, String msg, Integer delayTimeLevel) {
        try {
            Message message = new Message(topic, "", key, msg.getBytes(StandardCharsets.UTF_8));
            message.setDelayTimeLevel(delayTimeLevel);
            this.producer.send(message);
        } catch (MQClientException var7) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{},delayTimeLevel:{}", new Object[]{var7.getMessage(), topic, key, msg, delayTimeLevel});
        } catch (InterruptedException var8) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{},delayTimeLevel:{}", new Object[]{var8.getMessage(), topic, key, msg, delayTimeLevel});
        } catch (RemotingException var9) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{},delayTimeLevel:{}", new Object[]{var9.getMessage(), topic, key, msg, delayTimeLevel});
        } catch (MQBrokerException var10) {
            logger.error("ProducerService send() error,message:{},topic:{},key:{},msg:{},delayTimeLevel:{}", new Object[]{var10.getMessage(), topic, key, msg, delayTimeLevel});
        }
        
    }
    
    @PreDestroy
    public void shutDownProducer() {
        logger.info("ProducerService shutDownProducer is shutdown.");
        if (this.producer != null) {
            this.producer.shutdown();
        }
    }
}
